package com.okex.open.api.ws;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import com.okex.open.api.bean.other.OrderBookItem;
import com.okex.open.api.bean.other.SpotOrderBook;
import com.okex.open.api.bean.other.SpotOrderBookDiff;
import com.okex.open.api.bean.other.SpotOrderBookItem;
import com.okex.open.api.enums.CharsetEnum;
import com.okex.open.api.utils.DateUtils;
import lombok.Data;
import com.alibaba.fastjson.JSONObject;
import okhttp3.*;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.dromara.northstar.common.constant.ChannelType;
import org.dromara.northstar.common.constant.TickType;
import org.dromara.northstar.common.event.FastEventEngine;
import org.dromara.northstar.common.event.NorthstarEventType;
import org.dromara.northstar.common.model.core.Bar;
import org.dromara.northstar.common.model.core.Contract;
import org.dromara.northstar.common.model.core.Tick;
import org.dromara.northstar.gateway.IMarketCenter;

import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDate;
import java.time.LocalTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


public class BusinessWebSocketClient {

    //ws 模拟盘 the demo trading URL
    private static final String TEST_SERVICE_URL = "wss://wspap.okx.com:8443/ws/v5/public?brokerId=9999";

    //ws 实盘 the live trading URL
    private static final String SERVICE_URL = "wss://ws.okx.com:8443/ws/v5/public";

    //部分订阅频道已迁移
    private static final String BUSINESS_SERVICE_URL = "wss://ws.okx.com:8443/ws/v5/business";

    private static final String TEST_BUSINESS_SERVICE_URL = "wss://wspap.okx.com:8443/ws/v5/business?brokerId=9999";

    private static WebSocket webSocket = null;
    private static Boolean flag = false;
    private static Boolean isConnect = false;
    private static String sign;
    private final static HashFunction crc32 = Hashing.crc32();
    private final static ObjectReader objectReader = new ObjectMapper().readerFor(OrderBookData.class);
    private static Map<String, Optional<SpotOrderBook>> bookMap = new HashMap<>();
    public static Map<String, Contract> contractMap = new HashMap<>();
    private static String gatewayId;
    private static FastEventEngine feEngine = null;

    private static IMarketCenter mktCenter;

    public static volatile long lastUpdateTickTime = System.currentTimeMillis();


    public BusinessWebSocketClient(IMarketCenter mktCenter, FastEventEngine feEngine, String gatewayId, String apiKey, String secretKey, String passphrase, String accountType, String business) {
        this.mktCenter = mktCenter;
        this.feEngine = feEngine;
        this.gatewayId = gatewayId;
        if (business.equals("business")) {
            //与服务器建立连接
            connection(accountType.equals("1") ? TEST_SERVICE_URL : SERVICE_URL);
        } else {
            //与服务器建立连接
            connection(accountType.equals("1") ? TEST_BUSINESS_SERVICE_URL : BUSINESS_SERVICE_URL);
        }

        if (ObjectUtil.isEmpty(apiKey)) {
            //登录账号,用户需提供 api-key，passphrase,secret—key 不要随意透漏 ^_^
            login(apiKey, passphrase, secretKey);
        }
    }


    //与服务器建立连接，参数为服务器的URL
    public static WebSocket connection(final String url) {

        OkHttpClient client = new OkHttpClient.Builder()
                .readTimeout(5, TimeUnit.SECONDS)
                .build();
        Request request = new Request.Builder()
                .url(url)
                .build();

        webSocket = client.newWebSocket(request, new WebSocketListener() {
            ScheduledExecutorService service;

            @Override
            public void onOpen(final WebSocket webSocket, final Response response) {
                //连接成功后，设置定时器，每隔25s，自动向服务器发送心跳，保持与服务器连接
                isConnect = true;
                System.out.println(Instant.now().toString() + " Connected to the server success!");
                Runnable runnable = new Runnable() {
                    public void run() {
                        // task to run goes here
                        sendMessage("ping");
                    }
                };
                service = Executors.newSingleThreadScheduledExecutor();
                // 第二个参数为首次执行的延时时间，第三个参数为定时执行的间隔时间
                service.scheduleAtFixedRate(runnable, 25, 25, TimeUnit.SECONDS);
            }

            @Override
            public void onClosing(WebSocket webSocket, int code, String reason) {
                System.out.println("Connection is about to disconnect！");
                webSocket.close(1000, "Long time no message was sent or received！");
                webSocket = null;
            }

            @Override
            public void onClosed(final WebSocket webSocket, final int code, final String reason) {
                System.out.println("Connection dropped！");
            }

            @Override
            public void onFailure(final WebSocket webSocket, final Throwable t, final Response response) {
                System.out.println("Connection failed,Please reconnect!");
                System.out.println("reason: " + t.toString());
                if (Objects.nonNull(service)) {

                    service.shutdown();
                }
            }

            List<Double> bidPriceList = new ArrayList<>();
            List<Double> askPriceList = new ArrayList<>();
            List<Integer> bidVolumeList = new ArrayList<>();
            List<Integer> askVolumeList = new ArrayList<>();

            @Override
            public void onMessage(final WebSocket webSocket, final String bytes) {
                //测试服务器返回的字节
                 final String byteString = bytes.toString();


                //不进行解压
                final String s = byteString;
                if (s.contains("event")){
                    System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + " Receive: " + s);
                    return;
                }

                //判断是否是深度接口
                if (s.contains("\"channel\":\"books\",") || s.contains("\"channel\":\"books-l2-tbt\",") || s.contains("\"channel\":\"books50-l2-tbt\",")) {
                    //是深度接口

                    if (s.contains("snapshot")) {//记录下第一次的全量数据

                        JSONObject rst = JSONObject.parseObject(s);


                        JSONObject arg = rst.getJSONObject("arg");
                        JSONArray dataArr = rst.getJSONArray("data");
                        JSONObject data = dataArr.getJSONObject(0);
//
                        String dataStr = data.toString();

                        System.out.println("dataStr:" + dataStr);
                        Optional<SpotOrderBook> oldBook = parse(dataStr);
                        System.out.println("oldBook:" + oldBook);
                        String instrumentId = arg.get("instId").toString();
                        System.out.println("instrumentId:" + instrumentId);
                        bookMap.put(instrumentId, oldBook);
                    } else if (s.contains("\"action\":\"update\",")) {//是后续的增量，则需要进行深度合并


                        JSONObject rst = JSONObject.parseObject(s);
                        JSONObject arg = rst.getJSONObject("arg");
                        JSONArray dataArr = rst.getJSONArray("data");
                        JSONObject data = dataArr.getJSONObject(0);
                        String dataStr = data.toString();

                        String instrumentId = arg.get("instId").toString();

                        Optional<SpotOrderBook> oldBook = bookMap.get(instrumentId);
                        Optional<SpotOrderBook> newBook = parse(dataStr);

                        //获取增量的ask
                        List<SpotOrderBookItem> askList = newBook.get().getAsks();
                        //获取增量的bid
                        List<SpotOrderBookItem> bidList = newBook.get().getBids();

                        SpotOrderBookDiff bookdiff = oldBook.get().diff(newBook.get());


                        System.out.println("名称：" + instrumentId + ",深度合并成功！checknum值为：" + bookdiff.getChecksum() + ",合并后的数据为：" + bookdiff.toString());

                        String str = getStr(bookdiff.getAsks(), bookdiff.getBids());
                        System.out.println("名称：" + instrumentId + ",拆分要校验的字符串：" + str);
                        //计算checksum值
                        int checksum = checksum(bookdiff.getAsks(), bookdiff.getBids());
                        System.out.println("名称：" + instrumentId + ",校验的checksum:" + checksum);
                        boolean flag = checksum == bookdiff.getChecksum() ? true : false;
                        if (flag) {
                            System.out.println("名称：" + instrumentId + ",深度校验结果为：" + flag);
                            oldBook = parse(bookdiff.toString());
                            bookMap.put(instrumentId, oldBook);
                        } else {
                            System.out.println("名称：" + instrumentId + ",深度校验结果为：" + flag + "数据有误！请重连！");
                            //获取订阅的频道和币对
                            String channel = rst.get("table").toString();
                            String unSubStr = "{\"op\": \"unsubscribe\", \"args\":[\"" + channel + ":" + instrumentId + "\"]}";
                            System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + " Send: " + unSubStr);
                            webSocket.send(unSubStr);
                            String subStr = "{\"op\": \"subscribe\", \"args\":[\"" + channel + ":" + instrumentId + "\"]}";
                            System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + " Send: " + subStr);
                            webSocket.send(subStr);
                            System.out.println("名称：" + instrumentId + ",正在重新订阅！");
                        }
                    }
                } else if (s.contains("candle")) {
                    if (s.contains("subscribe")){
                        System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + " Receive: " + s);
                    }else {
                        JSONObject rst = JSONObject.parseObject(s);
                        JSONObject arg = rst.getJSONObject("arg");
                        String instId = arg.getString("instId");

                        Contract contract = contractMap.get(instId);
                        //成交量精度
                        double quantityPrecision = 1 / Math.pow(10, contract.quantityPrecision());
                        List<String[]> datas = JSON.parseArray(String.valueOf(rst.get("data")), String[].class);
                        for (String[] k : datas) {
                            //0：K线未完结
                            //1：K线已完结
                            if ("0".equals(k[8])) {
                                return;
                            }
                            Long actionTimestamp = Long.valueOf(k[0]);
                            Instant e = Instant.ofEpochMilli(actionTimestamp);
                            LocalTime actionTime = e.atZone(ZoneId.systemDefault()).toLocalTime();
                            LocalDate tradingDay = e.atZone(ZoneId.systemDefault()).toLocalDate();
                            double v = Double.parseDouble(k[6]) / quantityPrecision;
                            Double turnover = Double.parseDouble(k[7]);

                            Bar bar = Bar.builder()
                                    .contract(contract)
                                    .gatewayId(gatewayId)
                                    .tradingDay(tradingDay)
                                    .actionDay(tradingDay)
                                    .actionTime(actionTime)
                                    .actionTimestamp(actionTimestamp)
                                    .openPrice(Double.parseDouble(k[1]))
                                    .highPrice(Double.parseDouble(k[2]))
                                    .lowPrice(Double.parseDouble(k[3]))
                                    .closePrice(Double.parseDouble(k[4]))
                                    .volume((long) v)
                                    .volumeDelta((long) v)
                                    .turnover(turnover)
                                    .turnoverDelta(turnover)
                                    .channelType(ChannelType.OKX)
                                    .build();
                            feEngine.emitEvent(NorthstarEventType.BAR, bar);
                        }
                    }
                } else if (s.contains("pong")) {
                    System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + " Receive: " + s);

                } else if (null != s && s.contains("login")) {
                    System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + " Receive: " + s);
                    flag = true;
                } else {
                    //不是深度 k线接口
                    JSONObject rst = JSONObject.parseObject(s);
                    JSONObject arg = rst.getJSONObject("arg");
                    JSONArray dataArr = rst.getJSONArray("data");

                    String instId = arg.getString("instId");
                    JSONObject data = dataArr.getJSONObject(0);
                    Contract contract = contractMap.get(instId);

                    Long pushTimestamp = null;
                    Long localTimestamp = System.currentTimeMillis();
                    Long timing = null;

                    String channel = rst.getString("channel");
                    //深度
                    if (channel.contains("books")) {
                        // 从Map中获取买方和卖方的数据
                        List<List<String>> bids = (List<List<String>>) data.get("bids");
                        List<List<String>> asks = (List<List<String>>) data.get("asks");
                        bidPriceList.clear();
                        askPriceList.clear();
                        bidVolumeList.clear();
                        askVolumeList.clear();

                        // 分别提取买方和卖方的价格和数量
                        for (List<String> bid : bids) {
                            bidPriceList.add(Double.parseDouble(bid.get(0)));
                            bidVolumeList.add(Integer.valueOf(bid.get(1)));
                        }

                        for (List<String> ask : asks) {
                            askPriceList.add(Double.parseDouble(ask.get(0)));
                            askVolumeList.add(Integer.valueOf(ask.get(1)));
                        }
                    } else if (s.contains("tickers")) {
                        Long actionTimestamp = data.getLong("ts");
                        Instant e = Instant.ofEpochMilli(actionTimestamp);
                        LocalTime actionTime = e.atZone(ZoneId.systemDefault()).toLocalTime();
                        LocalDate tradingDay = e.atZone(ZoneId.systemDefault()).toLocalDate();

                        long Q = data.getLong("volCcy24h");
                        Tick tick = Tick.builder()
                                .contract(contract)
                                .tradingDay(tradingDay)
                                .actionDay(tradingDay)
                                .actionTime(actionTime)
                                .actionTimestamp(actionTimestamp)
                                .avgPrice(data.getDouble("w"))
                                .highPrice(data.getDouble("high24h"))
                                .lowPrice(data.getDouble("low24h"))
                                .openPrice(data.getDouble("open24h"))
                                .lastPrice(data.getDouble("last"))
                                //.settlePrice(isReasonable(upperLimit, lowerLimit, settlePrice) ? settlePrice : preSettlePrice)
                                //.openInterest(openInterest)
                                //.openInterestDelta(openInterestDelta)
                                .volume(Q)
                                .volumeDelta(Q)
                                //.turnover(turnover)
                                //.turnoverDelta(turnover)
                                //.lowerLimit(lowerLimit)
                                //.upperLimit(upperLimit)
                                //.preClosePrice(preClosePrice)
                                //.preSettlePrice(preSettlePrice)
                                //.preOpenInterest(preOpenInterest)
                                .askPrice(askPriceList)
                                .askVolume(askVolumeList)
                                .bidPrice(bidPriceList)
                                .bidVolume(bidVolumeList)
                                .gatewayId(gatewayId)
                                .channelType(ChannelType.OKX)
                                .type(TickType.MARKET_TICK)
                                .build();
                        mktCenter.onTick(tick);
                        feEngine.emitEvent(NorthstarEventType.TICK, tick);
                        lastUpdateTickTime = System.currentTimeMillis();
                    }
                    if (dataArr.toString().contains("\"ts\"")) {
                        pushTimestamp = Long.parseLong(data.get("ts").toString());

                        timing = localTimestamp - pushTimestamp;

                        System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + "(" + timing + "ms)" + " Receive: " + s);


                    } else {

                        System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + " Receive: " + s);
                    }


                }
//                }

            }
        });
        return webSocket;
    }


    private static void isLogin(String s) {
        if (null != s && s.contains("login")) {
            if (s.endsWith("true}")) {
                flag = true;
            }
        }
    }

    //获得sign
    private static String sha256_HMAC(String message, String secret) {
        String hash = "";
        try {
            Mac sha256_HMAC = Mac.getInstance("HmacSHA256");
            SecretKeySpec secret_key = new SecretKeySpec(secret.getBytes(CharsetEnum.UTF_8.charset()), "HmacSHA256");
            sha256_HMAC.init(secret_key);
            byte[] bytes = sha256_HMAC.doFinal(message.getBytes(CharsetEnum.UTF_8.charset()));
            hash = Base64.getEncoder().encodeToString(bytes);
        } catch (Exception e) {
            System.out.println("Error HmacSHA256 ===========" + e.getMessage());
        }
        return hash;
    }

    private static String listToJson(List<Map> list) {
        JSONArray jsonArray = new JSONArray();
        for (Map map : list) {
            jsonArray.add(JSONObject.parseObject(JSONObject.toJSONString(map)));
        }
        return JSONObject.toJSONString(jsonArray);
    }

    //登录
    public static void login(String apiKey, String passPhrase, String secretKey) {
        String timestamp = System.currentTimeMillis() / 1000 + "";
        String message = timestamp + "GET" + "/users/self/verify";
        sign = sha256_HMAC(message, secretKey);
        String str = "{\"op\"" + ":" + "\"login\"" + "," + "\"args\"" + ":" + "[{" + "\"apiKey\"" + ":" + "\"" + apiKey + "\"" + "," + "\"passphrase\"" + ":" + "\"" + passPhrase + "\"" + "," + "\"timestamp\"" + ":" + "\"" + timestamp + "\"" + "," + "\"sign\"" + ":" + "\"" + sign + "\"" + "}]}";
        sendMessage(str);
    }


    //订阅，参数为频道组成的集合
    public static void subscribe(List<Map> list) {
        String s = listToJson(list);
        String str = "{\"op\": \"subscribe\", \"args\":" + s + "}";
        if (null != webSocket)
            sendMessage(str);
    }

    //取消订阅，参数为频道组成的集合
    public static void unsubscribe(List<Map> list) {
        String s = listToJson(list);
        String str = "{\"op\": \"unsubscribe\", \"args\":" + s + "}";
        if (null != webSocket)
            sendMessage(str);
    }

    private static void sendMessage(String str) {
        if (null != webSocket) {
            try {
                Thread.sleep(1300);
            } catch (Exception e) {
                e.printStackTrace();
            }

            System.out.println(DateFormatUtils.format(new Date(), DateUtils.TIME_STYLE_S4) + "Send a message to the server:" + str);
            webSocket.send(str);
        } else {
            System.out.println("Please establish the connection before you operate it！");
        }
    }

    //断开连接
    public static void closeConnection() {
        if (null != webSocket) {
            webSocket.close(1000, "User actively closes the connection");
        } else {
            System.out.println("Please establish the connection before you operate it！");
        }
    }

    public boolean getIsLogin() {
        return flag;
    }

    public boolean getIsConnect() {
        return isConnect;
    }

    public static <T extends OrderBookItem> int checksum(List<T> asks, List<T> bids) {
        System.out.println("深度");
        StringBuilder s = new StringBuilder();
        for (int i = 0; i < 25; i++) {
            if (i < bids.size()) {
                s.append(bids.get(i).getPrice().toString());
                s.append(":");
                s.append(bids.get(i).getSize());
                s.append(":");
            }
            if (i < asks.size()) {
                s.append(asks.get(i).getPrice().toString());
                s.append(":");
                s.append(asks.get(i).getSize());
                s.append(":");
            }
        }
        final String str;
        if (s.length() > 0) {
            str = s.substring(0, s.length() - 1);
        } else {
            str = "";
        }

        return crc32.hashString(str, StandardCharsets.UTF_8).asInt();
    }

    private static <T extends OrderBookItem> String getStr(List<T> asks, List<T> bids) {
        StringBuilder s = new StringBuilder();
        for (int i = 0; i < 25; i++) {
            if (i < bids.size()) {
                s.append(bids.get(i).getPrice().toString());
                s.append(":");
                s.append(bids.get(i).getSize());
                s.append(":");
            }
            if (i < asks.size()) {
                s.append(asks.get(i).getPrice().toString());
                s.append(":");
                s.append(asks.get(i).getSize());
                s.append(":");
            }
        }
        final String str;
        if (s.length() > 0) {
            str = s.substring(0, s.length() - 1);
        } else {
            str = "";
        }
        return str;
    }

    public static Optional<SpotOrderBook> parse(String json) {


        try {
            OrderBookData data = objectReader.readValue(json);

            List<SpotOrderBookItem> asks =
                    data.getAsks().stream().map(x -> new SpotOrderBookItem(new String(x.get(0)), x.get(1), x.get(2), x.get(3)))
                            .collect(Collectors.toList());

            List<SpotOrderBookItem> bids =
                    data.getBids().stream().map(x -> new SpotOrderBookItem(new String(x.get(0)), x.get(1), x.get(2), x.get(3)))
                            .collect(Collectors.toList());
            return Optional.of(new SpotOrderBook(asks, bids, data.getTs(), data.getChecksum(), data.getPrevSeqId(), data.getSeqId()));
        } catch (Exception e) {
            System.out.println("e" + e);
            return Optional.empty();
        }
    }

    @Data
    public static class OrderBookData {
        private List<List<String>> asks;
        private List<List<String>> bids;
        private String ts;
        private int checksum;

        public long getPrevSeqId() {
            return prevSeqId;
        }

        public void setPrevSeqId(long prevSeqId) {
            this.prevSeqId = prevSeqId;
        }

        public long getSeqId() {
            return seqId;
        }

        public void setSeqId(long seqId) {
            this.seqId = seqId;
        }

        private long prevSeqId;
        private long seqId;

        public List<List<String>> getAsks() {
            return asks;
        }

        public void setAsks(List<List<String>> asks) {
            this.asks = asks;
        }

        public List<List<String>> getBids() {
            return bids;
        }

        public void setBids(List<List<String>> bids) {
            this.bids = bids;
        }

        public String getTs() {
            return ts;
        }

        public void setTs(String ts) {
            this.ts = ts;
        }

        public int getChecksum() {
            return checksum;
        }

        public void setChecksum(int checksum) {
            this.checksum = checksum;
        }
    }
}
